package com.zetyun.tiger.connector.read;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.zetyun.rt.meta.annotation.ActionMeta;
import com.zetyun.rt.sdk.action.MapAction;
import com.zetyun.rt.sdk.model.RtEvent;
import com.zetyun.rt.sdk.operator.OperatorContext;
import com.zetyun.tiger.connector.cdc.SimpleRedisContainer;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisSentinelConfig;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import static com.zetyun.tiger.connector.op.v2.RedisOpV2Action.build;

@ActionMeta(
        id = "tiger.read-redis",
        tags = {"redis"},
        category = "map",
        name = "Read-Redis",
        description = "Redis 读取 操作算子",
        parameterClass = ReadRedisParameter.class,
        status = ActionMeta.Status.PUBLIC
)
@Slf4j
public class ReadRedisAction extends MapAction {

    private SimpleRedisContainer rc;
    private ReadRedisParameter parameters;

    private ObjectMapper om;

    private List<ReadRedisBean> beans;

    @Override
    public void init(OperatorContext context) {
        super.init(context);


        parameters = context.getParameters();
        om = new ObjectMapper();
        beans = parameters.getBeans();
        FlinkJedisSentinelConfig conf =
                new FlinkJedisSentinelConfig.Builder()
                        .setMasterName(parameters.getMasterName())
                        .setSentinels(parameters.getSentinels())
                        .setPassword(parameters.getPassword())
                        .setDatabase(parameters.getDatabase())
                        .build();
        this.rc = build(conf);
        try {
            this.rc.open();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public RtEvent apply(RtEvent in) throws Exception {
        for (ReadRedisBean bean : beans) {
            readRedis(in, bean);
        }

        return in;
    }

    private RtEvent readRedis(RtEvent in, ReadRedisBean bean) throws IOException {
        String keyValue = in.getField(bean.getKeyName(), String.class);
        Object resValue = null;
        String additionalKey = bean.getAdditionalKey();
        Boolean needConcat = bean.getNeedConcat();
        if (needConcat) {
            String concatField = in.getField(bean.getConcatFieldName(), String.class);
            additionalKey = additionalKey.concat(concatField);
        }

        ReadRedisParameter.SimpleRedisCommand command = bean.getRc();

        if (keyValue == null || keyValue.isEmpty()) {
            in.setValue(bean.getResName(), resValue);
            return in;
        }

        keyValue = keyValue.trim();
        switch (bean.getRc()) {
            case HGET:
                resValue = rc.hget(additionalKey, keyValue);
                if (bean.getIsJson() && resValue != null) {
                    Map<String, Object> lastMap = om.readValue(resValue.toString(), new TypeReference<HashMap<String, Object>>() {
                    });
                    in.setValue(bean.getResName(), lastMap);
                } else {
                    in.setValue(bean.getResName(), resValue);
                }
                break;
            case GET:
                resValue = rc.get(keyValue);
                if (bean.getIsJson() && resValue != null) {
                    Map<String, Object> lastMap = om.readValue(resValue.toString(), new TypeReference<HashMap<String, Object>>() {
                    });
                    in.setValue(bean.getResName(), lastMap);
                } else {
                    in.setValue(bean.getResName(), resValue);
                }
                break;
            default:
                throw new IllegalArgumentException("暂不支持此类Redis命令: " + command);
        }

        return in;
    }

    @Override
    public void close() throws Exception {
        super.close();
        rc.close();
    }

}
